fix(sse): stop connection slots leaking when a client disconnects early - #532

Fixes #513. Independent of the other open PRs; branches off 0.16.1.

The leak

claim_sse_slot() runs in the handler; release_sse_slot() lived in the stream's finally.

An async generator that is never started never runs its finally. StreamingResponse.__call__ awaits stream_response, whose first statement is await send({"type": "http.response.start", ...}). If the client is already gone that raises ClientDisconnect before async for ever calls __anext__ -- so the generator body never executes and the slot is held for the life of the process.

Connect to /api/jobs/<id>/events, RST immediately, repeat 200 times: every progress stream and the queue stream answer 503 with zero live connections, until a restart. _sse_active has no ceiling reset and no reconciliation.

This does not need malice. A flaky network, or a page reloaded rapidly, leaks slots the same way.

Why not just claim inside the generator

That was the obvious fix and it does not work. By the time the generator runs, the response headers have gone out -- there is no status code left to send, so hitting the cap could no longer answer 503. The claim has to stay in the handler.

The fix

SseSlot keeps the claim in the handler and makes release idempotent, with __del__ as the backstop for the never-started case: collecting the generator collects the closure holding the slot. The stream still releases on its normal path, and the two cannot double-count.

queue.py takes the same guard, so both streams sharing the budget share the fix.

A bug the tests caught in the fix itself

_held is assigned before the claim:

def __init__(self) -> None:
    self._held = False
    claim_sse_slot()   # may raise 503
    self._held = True

__del__ runs on a half-built object too. Without that first line, a refused claim raised AttributeError out of __del__ instead of releasing nothing -- test_a_refused_claim_holds_nothing failed on the first run and found it.

Verification

New tests/test_sse_slot_budget.py, 5 tests. Confirmed not vacuous -- removing the __del__ backstop fails test_a_slot_dropped_without_release_is_reclaimed.

ruff check       All checks passed
ruff format      94 files already formatted
pytest tests/    898 passed, 2 failed

The 2 failures are the pre-existing ogg pair, which fail identically on 0.16.1.

One caveat for reviewers

_sse_active is a plain module-level int with no lock. That predates this change, and everything normally touches it from the event loop -- but __del__ can run on whichever thread triggers collection, so the backstop is a slightly wider surface than the existing code had. Given the value only ever moves by one and the alternative was a permanent leak, that felt like the right trade; worth a second opinion. A lock around both helpers would close it if you want belt and braces.